package com.imooc.structured.streaming

import java.sql.Timestamp

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions._

/**
  * @description
  * @author yuyon26@126.com
  * @date 2018/10/27 21:03
  */
object StructuredNetworkWordCountWindowed {
  def main(args: Array[String]): Unit = {

    val spark = SparkSession
      .builder
      .appName("StructuredNetworkWordCountWindowed")
      .master("local[2]")
      .getOrCreate()
    spark.sparkContext.setLogLevel("ERROR")

    import spark.implicits._
    //创建一个DataFrame
    val linesDataFrames = spark.readStream
      .format("socket")
      .option("host", "192.168.31.30")
      .option("port", 9999)
      .option("includeTimestamp",true)
      .load()

    //将行拆分为单词，保留时间戳
    val words = linesDataFrames.as[(String, Timestamp)].flatMap(line =>
      line._1.split(" ").map(word => (word, line._2))
    ).toDF("word", "timestamp")

    //按窗口和单词对数据进行分组，并计算每个组的计数
    val windowedCounts = words.groupBy(
      window($"timestamp", "10 seconds", "5 seconds"), $"word"
    ).count().orderBy("window")

    //开始运行将窗口化的单词计数打印到控制台的查询
    val query = windowedCounts.writeStream
      .outputMode("complete")
      .format("console")
      .option("truncate", "false")
      .start()

    query.awaitTermination()

  }
}
